#include "stdafx.h"
#include "TQSManager.h"
#include "IOCompletionPort.h"
#include "ThreadIOCP.h"
#include "DBFCircleDequeue.h"
#include "ThreadReadFile.h"
#include "DataDefinition.h"
#include "FixCircleDequeue.h"
#include "ThreadHeartBeat.h"
#include "../DzhNewTQSDlg/ConfigInfoInterface.h"
#include <vector>

const int WORKER_THREADS_PER_PROCESSOR = 2;

LPCompress g_funCompress = NULL;
LPUncompress g_funUncompress = NULL;

TQSManager::TQSManager()
	:m_pcMemPool(NULL)
	,m_clsClientInfo()
{
	m_pcMemPool = MemoryPool::Instance();
}

TQSManager::~TQSManager()
{
	if(NULL != m_pcMemPool)
	{
		delete m_pcMemPool;
		m_pcMemPool = NULL;
	}
}

void TQSManager::StartServer()
{
	// Initialize circle dequeue
	map<int,CircleDequeueBase*> mapCircleQueue;

	if (false == LoadDll())
	{
		LOG(ERROR) << "Error for loading dll";
		return;
	}
	
	// Add circle queue
	if(false == AddCircleQueue(mapCircleQueue))
	{
		map<int,CircleDequeueBase*>::iterator it = mapCircleQueue.begin();
		while(it != mapCircleQueue.end())
		{
			delete it->second;
			it++;
		}

		LOG(ERROR) << "Error to add circle queue";
		return;
	}
	
	// Initialize IOCP instance
	ServerSet stServer = CConfigInfoInterface::Instance()->GetServerSetting();
	IOCompletionPort* pIOCP = IOCompletionPort::Instance(stServer.m_strHostIP,stServer.m_uiHostPort,10);
	if(NULL == pIOCP)
	{
		LOG(ERROR) << "Cannot get IOCP instance";
		return;
	}

	pIOCP->InitializeListenSocket();

	// Get processor number and calculate threads number
	SYSTEM_INFO si;
	GetSystemInfo(&si);
	int iThreadCount = si.dwNumberOfProcessors * WORKER_THREADS_PER_PROCESSOR + 1;

	vector<ThreadIOCP*> vecIOCPThread;
	for(int i = 0; i < iThreadCount; i++)
	{
		ThreadIOCP* pIOCPThread = new ThreadIOCP(mapCircleQueue,&m_clsClientInfo);
		vecIOCPThread.push_back(pIOCPThread);
		map<int,CircleDequeueBase*>::iterator it = mapCircleQueue.begin();
		while(it != mapCircleQueue.end())
		{
			it->second->Attach(pIOCPThread);
			it++;
		}

		pIOCPThread->BeginThread();
	}
	LOG(INFO) << "IOCP threads have started up";

	// Initialize DBF receive thread
	ThreadReadFile* pcFileProcessThread = new ThreadReadFile(mapCircleQueue);
	if(pcFileProcessThread == NULL)
	{
		LOG(FATAL) << "Cannot new read dbf thread instance";
		return;
	}

	pcFileProcessThread->BeginThread();
	LOG(INFO) << "DBF thread has started up";

	ThreadHeartBeat* pclsHeartBeatThread = new ThreadHeartBeat(&m_clsClientInfo,mapCircleQueue);
	if(pcFileProcessThread == NULL)
	{
		LOG(FATAL) << "Cannot new heart beat thread instance";
		return;
	}
	
	pclsHeartBeatThread->BeginThread();
	LOG(INFO) << "Heart beat thread has started up";

	while(1)
	{
		Sleep(5000);
	}
}

map<SOCKET,Connection> TQSManager::GetClientInformation()
{
	return m_clsClientInfo.GetConnectionList();
}

bool TQSManager::IsFileChecked(const int p_iFileIndex)
{
	vector<HqInfoSet> vecHqInfo = CConfigInfoInterface::Instance()->GetHqInfoSetting();
	vector<HqInfoSet>::iterator it = vecHqInfo.begin();
	while (it != vecHqInfo.end())
	{
		if (it->m_uiID == p_iFileIndex && it->m_bIsChecked == true)
		{
			return true;
		}

		it++;
	}
	
	return false;
}

bool TQSManager::AddCircleQueue(map<int,CircleDequeueBase*> &p_mapCircleQueue)
{
	bool bRet = true;

	if (true == IsFileChecked(FILETYPE_GZLX))
	{
		// Gzlx
		bRet = AddCircleQueueMap(p_mapCircleQueue,1024*1024,10,1024*1024,GZLX_BEGIN_ID,GZLX_END_ID,FILETYPE_GZLX,FILETYPE_GZLX_INCR,0);
		if (false == bRet)
		{
			return false;
		}
	}

	if (true == IsFileChecked(FILETYPE_STBHQ))
	{
		// STBHQ
		bRet = AddCircleQueueMap(p_mapCircleQueue,1024*1024,10,1024*1024,STBHQ_BEGIN_ID,STBHQ_END_ID,FILETYPE_STBHQ,FILETYPE_STBHQ_INCR,0);
		if (false == bRet)
		{
			return false;
		}
	}

	if (true == IsFileChecked(FILETYPE_50ETF))
	{
		// 50Etf
		bRet = AddCircleQueueMap(p_mapCircleQueue,1024*1024,1,1024*10,ETF_BEGIN_ID,ETF_END_ID,FILETYPE_50ETF,FILETYPE_50ETF,1);
		if (false == bRet)
		{
			return false;
		}
	}

	if (true == IsFileChecked(FILETYPE_SJSXX))
	{
		// SJSXX
		bRet = AddCircleQueueMap(p_mapCircleQueue,1024*1024,10,1024*1024,SJSXX_BEGIN_ID,SJSXX_END_ID,FILETYPE_SJSXX,FILETYPE_SJSXX,0);
		if (false == bRet)
		{
			return false;
		}
	}

	if (true == IsFileChecked(FILETYPE_SHOW))
	{
		// Show2003
		bRet = AddCircleQueueMap(p_mapCircleQueue,5*1024*1024,20,1024*1024*2,SHOW2003_BEGIN_ID,SHOW2003_END_ID,FILETYPE_SHOW,FILETYPE_SHOW_INCR,0);
		if (false == bRet)
		{
			return false;
		}
	}

	if (true == IsFileChecked(FILETYPE_SJSHQ))
	{
		// Sjshq
		bRet = AddCircleQueueMap(p_mapCircleQueue,5*1024*1024,20,1024*1024*2,SJSHQ_BEGIN_ID,SJSHQ_END_ID,FILETYPE_SJSHQ,FILETYPE_SJSHQ_INCR,0);
		if (false == bRet)
		{
			return false;
		}
	}

	if (true == IsFileChecked(FILETYPE_ZHB))
	{
		// ZHB
		bRet = AddCircleQueueMap(p_mapCircleQueue,5*1024*1024,10,1024*1024,ZHB_BEGIN_ID,ZHB_END_ID,FILETYPE_ZHB,FILETYPE_ZHB_INCR,0);
		if (false == bRet)
		{
			return false;
		}
	}

	if (true == IsFileChecked(FILETYPE_ZHGGQZ))
	{
		// ZHGGQZ
		bRet = AddCircleQueueMap(p_mapCircleQueue,5*1024*1024,10,1024*1024,ZHGGQZ_BEGIN_ID,ZHGGQZ_END_ID,FILETYPE_ZHGGQZ,FILETYPE_ZHGGQZ_INCR,0);
		if (false == bRet)
		{
			return false;
		}
	}

	if (true == IsFileChecked(FILETYPE_ZHJBXX))
	{
		// ZHJBXX
		bRet = AddCircleQueueMap(p_mapCircleQueue,5*1024*1024,10,1024*1024,ZHJBXX_BEGIN_ID,ZHJBXX_END_ID,FILETYPE_ZHJBXX,FILETYPE_ZHJBXX_INCR,0);
		if (false == bRet)
		{
			return false;
		}
	}

	if (true == IsFileChecked(FILETYPE_HQZ))
	{
		// HQZ
		bRet = AddCircleQueueMap(p_mapCircleQueue,5*1024*1024,1,1024*1024,HQZ_BEGIN_ID,HQZ_END_ID,FILETYPE_HQZ,FILETYPE_HQZ,1);
		if (false == bRet)
		{
			return false;
		}
	}

	if (true == IsFileChecked(FILETYPE_HTIDXCON))
	{
		// HTIDXCON
		bRet = AddCircleQueueMap(p_mapCircleQueue,5*1024*1024,1,1024*1024,HTIDXCON_BEGIN_ID,HTIDXCON_END_ID,FILETYPE_HTIDXCON,FILETYPE_HTIDXCON,1);
		if (false == bRet)
		{
			return false;
		}
	}

	if (true == IsFileChecked(FILETYPE_ZXZS))
	{
		// ZXZS
		bRet = AddCircleQueueMap(p_mapCircleQueue,5*1024*1024,10,1024*1024,ZXZS_BEGIN_ID,ZXZS_END_ID,FILETYPE_ZXZS,FILETYPE_ZXZS_INCR,0);
		if (false == bRet)
		{
			return false;
		}
	}

	if (true == IsFileChecked(FILETYPE_HTINDEX))
	{
		// HTINDEX
		bRet = AddCircleQueueMap(p_mapCircleQueue,5*1024*1024,10,1024*1024,HTINDEX_BEGIN_ID,HTINDEX_END_ID,FILETYPE_HTINDEX,FILETYPE_HTINDEX_INCR,0);
		if (false == bRet)
		{
			return false;
		}
	}

	if (true == IsFileChecked(FILETYPE_HGXXN))
	{
		// HGXXN
		bRet = AddCircleQueueMap(p_mapCircleQueue,5*1024*1024,10,1024*1024,HGXXN_BEGIN_ID,HGXXN_END_ID,FILETYPE_HGXXN,FILETYPE_HGXXN_INCR,0);
		if (false == bRet)
		{
			return false;
		}
	}

	if (true == IsFileChecked(FILETYPE_CPXX))
	{
		// CPXX
		bRet = AddCircleQueueMap(p_mapCircleQueue,5*1024*1024,1,1024*1024,CPXX_BEGIN_ID,CPXX_END_ID,FILETYPE_CPXX,FILETYPE_CPXX,1);
		if (false == bRet)
		{
			return false;
		}
	}

	if (true == IsFileChecked(FILETYPE_HGHQ))
	{
		// HGHQ
		bRet = AddCircleQueueMap(p_mapCircleQueue,5*1024*1024,10,1024*1024,HGHQ_BEGIN_ID,HGHQ_END_ID,FILETYPE_HGHQ,FILETYPE_HGHQ_INCR,0);
		if (false == bRet)
		{
			return false;
		}
	}

	return bRet;
}

bool TQSManager::AddCircleQueueMap(map<int,CircleDequeueBase*> &p_mapCircleQueue,
	unsigned int p_uiEntireBufferSize,
	unsigned int p_uiElementAmount, 
	unsigned int p_uiElementSize,
	unsigned int p_uiBeginSequence,
	unsigned int p_uiEndSequence,
	unsigned char p_uiEntireDataType, 
	unsigned char p_uiIncrementDataType,
	unsigned char p_ucCircleQueueType)
{
	CircleDequeueBase* pcCircleQueue = NULL;
	switch(p_ucCircleQueueType)
	{
	case 0:	// DBF file
		pcCircleQueue = new DBFCircleDequeue(p_uiEntireBufferSize,p_uiElementAmount,p_uiElementSize,p_uiBeginSequence,p_uiEndSequence,p_uiEntireDataType,p_uiIncrementDataType);
		break;
	case 1: // FIX file
		pcCircleQueue = new FixCircleDequeue(p_uiEntireBufferSize,p_uiElementAmount,p_uiElementSize,p_uiBeginSequence,p_uiEndSequence,p_uiEntireDataType,p_uiIncrementDataType);
		break;
	default:
		break;
	}

	if(pcCircleQueue == NULL)
	{
		LOG(FATAL) << "pcCircleQueue is NULL";
		return false;
	}
	if(false == pcCircleQueue->Initialize())
	{
		LOG(FATAL) << "pcCircleQueue->Initialize() is false";
		return false;
	}

	p_mapCircleQueue.insert(pair<int,CircleDequeueBase*>(p_uiEntireDataType,pcCircleQueue));
	return true;
}

bool TQSManager::LoadDll()
{
	HINSTANCE m_hCompressDll = LoadLibrary((LPCWSTR)L"./Debug/zlib.dll");
	if (!m_hCompressDll)
	{
		LOG(ERROR) << "Cannot get instance of zlib library";
		return false;
	}

	g_funCompress = (LPCompress)GetProcAddress(m_hCompressDll,"compress2");
	g_funUncompress =(LPUncompress)GetProcAddress(m_hCompressDll,"uncompress");
	if (!g_funCompress || !g_funUncompress)
	{
		LOG(ERROR) << "Cannot load dll";
		return false;
	}

	return true;
}